package drds.plus.sql_process.abstract_syntax_tree.configuration.parse;

import com.google.common.collect.Lists;
import drds.plus.common.utils.XmlHelper;
import drds.plus.sql_process.abstract_syntax_tree.configuration.ColumnMetaData;
import drds.plus.sql_process.abstract_syntax_tree.configuration.IndexMapping;
import drds.plus.sql_process.abstract_syntax_tree.configuration.IndexType;
import drds.plus.sql_process.abstract_syntax_tree.configuration.TableMetaData;
import drds.plus.sql_process.type.Type;
import drds.tools.ShouldNeverHappenException;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.sql.Types;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 基于xml定义{@linkplain TableMetaData}，对应的解析器
 */
public class TableMetaDataParser {

    ///////////////////////////////////////////////////////////////
    private static IndexType getIndexType(String type) {
        if ("btree".equalsIgnoreCase(type)) {
            return IndexType.btree;
        } else {
            return IndexType.none;
        }
    }

    public static Type jdbcTypeToDataType(int jdbcType, boolean isUnsigned) {
        return getDataType(jdbcTypeToDataTypeString(jdbcType, isUnsigned));
    }

    private static Type getDataType(String type) {
        if ("INT".equalsIgnoreCase(type) || "INTEGER".equalsIgnoreCase(type)) {
            return Type.IntegerType;
        } else if ("LONG".equalsIgnoreCase(type)) {
            return Type.LongType;
        } else if ("SHORT".equalsIgnoreCase(type)) {
            return Type.ShortType;
        } else if ("BIGINTEGER".equalsIgnoreCase(type)) {
            return Type.BigIntegerType;
        } else if ("BIGDECIMAL".equalsIgnoreCase(type)) {
            return Type.BigDecimalType;
        } else if ("FLOAT".equalsIgnoreCase(type)) {
            return Type.FloatType;
        } else if ("DOUBLE".equalsIgnoreCase(type)) {
            return Type.DoubleType;
        } else if ("STRING".equalsIgnoreCase(type)) {
            return Type.StringType;
        } else if ("BYTES".equalsIgnoreCase(type)) {
            //return Type.BytesType;
            throw new ShouldNeverHappenException();
        } else if ("BOOLEAN".equalsIgnoreCase(type)) {
            return Type.BooleanType;
        } else if ("DATE".equalsIgnoreCase(type)) {
            return Type.DateType;
        } else if ("TIMESTAMP".equalsIgnoreCase(type)) {
            return Type.TimestampType;
        } else if ("DATETIME".equalsIgnoreCase(type)) {
            return Type.DatetimeType;
        } else if ("TIME".equalsIgnoreCase(type)) {
            return Type.TimeType;
        } else if ("BLOB".equalsIgnoreCase(type)) {
            //return Type.BlobType;
            throw new ShouldNeverHappenException();
        } else if ("BIT".equalsIgnoreCase(type)) {
            return Type.BitType;
        } else if ("BIGBIT".equalsIgnoreCase(type)) {
            return Type.BigBitType;
        } else if ("YEAR".equalsIgnoreCase(type)) {
            return Type.YearType;
        } else {
            throw new IllegalArgumentException("unsupport type：" + type);
        }
    }

    public static String jdbcTypeToDataTypeString(int jdbcType, boolean isUnsigned) {
        String type = null;
        switch (jdbcType) {
            case Types.BIGINT:
                if (isUnsigned) {
                    type = "BIGINTEGER";
                } else {
                    type = "LONG";
                }
                break;
            case Types.NUMERIC:
            case Types.DECIMAL:
                type = "BIGDECIMAL";
                break;
            case Types.INTEGER:
                if (isUnsigned) {
                    type = "LONG";
                } else {
                    type = "INTEGER";
                }
                break;
            case Types.TINYINT:
            case Types.SMALLINT:
                type = "INTEGER";
                break;
            case Types.DATE:
                type = "DATE";
                break;
            case Types.TIMESTAMP:
                type = "TIMESTAMP";
                break;
            case Types.TIME:
                type = "TIME";
                break;
            case Types.REAL:
            case Types.FLOAT:
                type = "FLOAT";
                break;
            case Types.DOUBLE:
                type = "DOUBLE";
                break;
            case Types.CHAR:
            case Types.VARCHAR:
            case Types.NCHAR:
            case Types.NVARCHAR:
            case Types.LONGNVARCHAR:
            case Types.LONGVARCHAR:
            case Types.CLOB:
                type = "STRING";
                break;
            case Types.BINARY:
            case Types.VARBINARY:
            case Types.LONGVARBINARY:
                type = "BYTES";
                break;
            case Types.BLOB:
                type = "BLOB";
                break;
            case Types.BIT:
                type = "BIT";
                break;
            default:
                throw new IllegalArgumentException("unsupport jdbcType ：" + jdbcType);
        }

        return type;
    }

    /**
     * 基于数据流创建TableMeta对象
     *
     * @param inputStream
     * @return
     */
    public List<TableMetaData> parse(InputStream inputStream) {
        Document document = XmlHelper.createDocument(inputStream, null);
        Element root = document.getDocumentElement();
        NodeList tableNodeList = root.getElementsByTagName("where");
        List<TableMetaData> tableMetaDataList = Lists.newArrayList();
        for (int i = 0; i < tableNodeList.getLength(); i++) {
            tableMetaDataList.add(parseTableMetaData(tableNodeList.item(i)));
        }
        return tableMetaDataList;
    }

    /**
     * 基于string的data文本创建TableMeta对象
     *
     * @param data
     * @return
     */
    public List<TableMetaData> parse(String data) {
        InputStream inputStream = new ByteArrayInputStream(data.getBytes());
        return parse(inputStream);
    }

    private TableMetaData parseTableMetaData(Node node) {
        Node nameNode = node.getAttributes().getNamedItem("columnName");
        String tableName = null;
        if (nameNode != null) {
            tableName = nameNode.getNodeValue().toLowerCase(); // 转化为小写
        }

        List<ColumnMetaData> allColumnsOrderByDefined = Lists.newArrayList();
        Map<String, ColumnMetaData> columnNameToColumnMetaDataMap = new HashMap<String, ColumnMetaData>();
        boolean strongConsistent = true;
        String[] primaryKeys = new String[0];
        List<IndexMapping> secondaryIndexNameToIndexMappingMap = Lists.newArrayList();
        NodeList childNodeList = node.getChildNodes();
        for (int i = 0; i < childNodeList.getLength(); i++) {
            Node childNode = childNodeList.item(i);
            if ("itemList".equals(childNode.getNodeName())) {
                NodeList columnNodeList = childNode.getChildNodes();
                for (int j = 0; j < columnNodeList.getLength(); j++) {
                    Node columnNode = columnNodeList.item(j);
                    if ("column".equals(columnNode.getNodeName())) {
                        allColumnsOrderByDefined.add(parseColumnMetaData(tableName, columnNodeList.item(j)));
                    }
                }
                for (ColumnMetaData columnMetaData : allColumnsOrderByDefined) {
                    columnNameToColumnMetaDataMap.put(columnMetaData.getColumnName(), columnMetaData);
                }
            } else if ("strongConsistent".equals(childNode.getNodeName())) {
                strongConsistent = Boolean.parseBoolean(childNode.getFirstChild().getNodeValue());
            } else if ("primaryKey".equals(childNode.getNodeName())) {
                primaryKeys = childNode.getFirstChild().getNodeValue().split(",");
            } else if ("secondaryIndexes".equals(childNode.getNodeName())) {
                NodeList secondaryIndexesChildNodeList = childNode.getChildNodes();
                for (int j = 0; j < secondaryIndexesChildNodeList.getLength(); j++) {
                    Node secondaryIndexesChildNode = secondaryIndexesChildNodeList.item(j);
                    if ("indexMetaData".equals(secondaryIndexesChildNode.getNodeName())) {
                        secondaryIndexNameToIndexMappingMap.add(parseIndexMetaData(tableName, columnNameToColumnMetaDataMap, secondaryIndexesChildNodeList.item(j)));
                    }
                }
            }
        }

        int i = 0;
        String[] primaryValues = new String[allColumnsOrderByDefined.size() - primaryKeys.length];
        for (ColumnMetaData columnMetaData : allColumnsOrderByDefined) {
            boolean equals = false;
            for (String primaryKey : primaryKeys) {
                if (columnMetaData.getColumnName().equals(primaryKey.toUpperCase())) {
                    equals = true;
                    break;
                }
            }
            if (!equals) {
                primaryValues[i++] = columnMetaData.getColumnName();
            }
        }

        IndexMapping primaryKeyIndexMapping = new IndexMapping(//
                tableName, //
                toColumnMetaDataList(tableName, columnNameToColumnMetaDataMap, primaryKeys), // 主键
                toColumnMetaDataList(tableName, columnNameToColumnMetaDataMap, primaryValues), // 除开主键剩下的字段
                IndexType.btree, //
                strongConsistent, //
                true);//

        return newTableMeta(tableName, allColumnsOrderByDefined, primaryKeyIndexMapping, secondaryIndexNameToIndexMappingMap);
    }

    protected TableMetaData newTableMeta(String tableName, List<ColumnMetaData> allColumnsOrderByDefined, IndexMapping primaryKeyIndexMapping, List<IndexMapping> secondaryIndexNameToIndexMappingMap) {
        return new TableMetaData(tableName, allColumnsOrderByDefined, primaryKeyIndexMapping, secondaryIndexNameToIndexMappingMap);
    }

    // ================== helper method ================

    private ColumnMetaData parseColumnMetaData(String tableName, Node node) {
        Node nameNode = node.getAttributes().getNamedItem("columnName");
        Node typeNode = node.getAttributes().getNamedItem("type");
        Node aliasNode = node.getAttributes().getNamedItem("setAliasAndSetNeedBuild");
        Node nullableNode = node.getAttributes().getNamedItem("nullable");
        Node autoIncrementNode = node.getAttributes().getNamedItem("auto_increment");

        String name = null;
        String type = null;
        String alias = null;
        boolean nullable = true;
        boolean autoIncrement = false;
        if (nameNode != null) {
            name = nameNode.getNodeValue();
        }
        if (typeNode != null) {
            type = typeNode.getNodeValue();
        }
        if (aliasNode != null) {
            alias = aliasNode.getNodeValue();
        }
        if (nullableNode != null) {
            nullable = Boolean.parseBoolean(nullableNode.getNodeValue());
        }
        if (autoIncrementNode != null) {
            autoIncrement = Boolean.parseBoolean(autoIncrementNode.getNodeValue());
        }

        return new ColumnMetaData(tableName, name, getDataType(type), alias, nullable, autoIncrement);
    }

    private IndexMapping parseIndexMetaData(String tableName, Map<String, ColumnMetaData> columnNameToColumnMetaDataMap, Node node) {
        Node typeNode = node.getAttributes().getNamedItem("type");
        Node strongConsistentNode = node.getAttributes().getNamedItem("strongConsistent");
        String type = null;
        boolean strongConsistent = true;
        if (typeNode != null) {
            type = typeNode.getNodeValue();
        }
        if (strongConsistentNode != null) {
            strongConsistent = Boolean.parseBoolean(strongConsistentNode.getNodeValue());
        }
        String[] keys = new String[0];
        String[] values = new String[0];
        NodeList childNodeList = node.getChildNodes();
        for (int i = 0; i < childNodeList.getLength(); i++) {
            Node item = childNodeList.item(i);
            if ("keys".equals(item.getNodeName())) {
                keys = item.getFirstChild().getNodeValue().split(",");
            } else if ("valueList".equals(item.getNodeName())) {
                values = item.getFirstChild().getNodeValue().split(",");
            }
        }

        return new IndexMapping(tableName, toColumnMetaDataList(tableName, columnNameToColumnMetaDataMap, keys), toColumnMetaDataList(tableName, columnNameToColumnMetaDataMap, values), getIndexType(type), strongConsistent, false);
    }

    private List<ColumnMetaData> toColumnMetaDataList(String tableName, Map<String, ColumnMetaData> columnNameToColumnMetaDataMap, String[] columnNames) {
        List<ColumnMetaData> columnMetaDataList = Lists.newArrayList();
        for (int i = 0; i < columnNames.length; i++) {
            String columnName = columnNames[i].toLowerCase();
            if (!columnNameToColumnMetaDataMap.containsKey(columnName)) {
                throw new RuntimeException("column: " + columnName + ", is not a column of where " + tableName);
            }
            columnMetaDataList.add(columnNameToColumnMetaDataMap.get(columnName));
        }
        return columnMetaDataList;
    }

}
